package com.raymond.client.connector;


import com.raymond.client.config.SocketConfig;
import com.raymond.client.session.AbstractClientSession;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 客户端连接管理
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2022-10-26 17:49
 */
public abstract class AbstractClientConnector implements IConnector {
    private final static Logger logger = LoggerFactory.getLogger(AbstractClientConnector.class);

    private Bootstrap clientBootstrap;
    private EventLoopGroup workerGroup;
    private DefaultEventExecutorGroup defaultEventExecutorGroup;


    public AbstractClientConnector() {
        this(null);
    }

    public AbstractClientConnector(EventLoopGroup workerGroup) {
        //提供默认workerGroup
        if (workerGroup == null) {
            workerGroup = new NioEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet()));
                }
            });
        }
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
                1,
                new ThreadFactory() {
                    private AtomicInteger threadIndex = new AtomicInteger(0);
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "NettyClientSessionThread_" + this.threadIndex.incrementAndGet());
                    }
                });
        this.workerGroup = workerGroup;
        this.clientBootstrap = new Bootstrap();
        clientBootstrap.group(this.workerGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_RCVBUF, 16384).option(ChannelOption.SO_SNDBUF, 8192)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(1024))
                .handler(buildChannelInitializer());;
    }

    /**
     * 绑定
     * @return
     */
    protected abstract ChannelInitializer<?> buildChannelInitializer();


    @Override
    public void bind(AbstractClientSession clientSession) {
        SocketConfig socketConfig = clientSession.getSocketConfig();
        this.clientBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) socketConfig.getConnectTimeout());
        // attempt to connect to the remote system
        InetSocketAddress socketAddress = new InetSocketAddress(socketConfig.getHost(), socketConfig.getPort());
        ChannelFuture connectFuture = this.clientBootstrap.connect(socketAddress).addListener((ChannelFutureListener) future -> {
            future.channel().pipeline().addLast(defaultEventExecutorGroup, clientSession.getSessionHandler());
            if (!future.isSuccess()) {
                future.channel().pipeline().fireChannelInactive();
            }
        });
        logger.debug("attempt to connect to the remote system host:{}, port:{}", socketAddress.getHostString(), socketAddress.getPort());
    }

    @Override
    public void destroy() {
        this.workerGroup.shutdownGracefully();
        this.defaultEventExecutorGroup.shutdownGracefully();
    }


}
